import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple6;

public class CrosswithProjectionTxt
{


    public static void main(String[] args) throws Exception
    {


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String rootPath = new StringBuilder(System.getProperty("user.dir")).toString();
        DataSet<String> s_students = env.readTextFile("file:///"+rootPath+"/"+"students.csv");
        DataSet<String> s_scores = env.readTextFile("file:///"+rootPath+"/"+"scores.csv");

        DataSet<Tuple3<Integer,String,String>> students  =s_students.map(new MapFunction<String, Tuple3<Integer, String, String>>() {
                    @Override
                    public Tuple3<Integer, String, String> map(String s) throws Exception {
                        String[] line = s.split(" ");
                        return new Tuple3<Integer, String, String>(Integer.valueOf(line[0]), line[1], line[2]);
                    }
                } );


        DataSet<Tuple4<Integer, Integer, Integer, Integer>> scores = s_scores.map(new MapFunction<String, Tuple4<Integer, Integer, Integer, Integer>>() {

            @Override
            public Tuple4<Integer, Integer, Integer, Integer> map(String s) throws Exception {
                String[] line = s.split(" ");
                return new Tuple4<Integer, Integer, Integer, Integer>(Integer.valueOf(line[0]), Integer.valueOf(line[1]),
                        Integer.valueOf(line[2]), Integer.valueOf(line[3]));
            }
        });

//        DataSet<Tuple4<Integer, Byte, Integer, Double> > result0 = input1.cross(input2);

        DataSet<Tuple6<Integer, String, String, Integer, Integer, Integer>>  result = students.join(scores).where(0).equalTo(0)
//                 select and reorder fields of matching tuples
                .projectFirst(0,1,2).projectSecond(1,2,3);
//        这些projectSecond和projectFirst的作用是:
//        把结果打散，然后排序
        result.print();

//没有projection的效果
//        ((1,111,yuchi),(1,3.4))
//        ((1,111,yuchi),(2,3.5))
//        ((1,111,yuchi),(3,3.4))
//        ((1,111,yuchi),(2,3.5))
//        ((2,87,chiyu),(1,3.4))
//        ((2,87,chiyu),(2,3.5))
//        ((2,87,chiyu),(3,3.4))
//        ((2,87,chiyu),(2,3.5))


//     有projection的效果:
//        (3,111,1,3.4)
//        (2,111,1,3.5)
//        (1,111,1,3.4)
//        (2,111,1,3.5)
//        (3,87,2,3.4)
//        (2,87,2,3.5)
//        (1,87,2,3.4)
//        (2,87,2,3.5)


    }

}
